package com.zhss.dfs.server;

import com.alibaba.fastjson.JSONArray;
import com.zhss.rpc.model.NameNodeModel;
import com.zhss.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;

import java.io.File;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 负责跟一组NameNode中的某一个进行通信的线程组件
 *
 * @author zhonghuashishan
 */
public class NameNodeServiceActor {

    private final ManagedChannel channel;
    private final NameNodeServiceGrpc.NameNodeServiceBlockingStub blockingStub;

    // 构造函数，初始化 gRPC 通道和服务存根
    public NameNodeServiceActor(String host, int port) {
        this.channel = ManagedChannelBuilder.forAddress(host, port)
                .usePlaintext()
                .build();
        this.blockingStub = NameNodeServiceGrpc.newBlockingStub(channel);
    }

    /**
     * 向nameNode上报自身的文件对应的data节点
     */
    public void InformReceivedReplicateInfo(String fileName, String hostname, String ip, Long fileSize) {
        NameNodeModel.InformReceivedReplicateRequest request = NameNodeModel.InformReceivedReplicateRequest.newBuilder()
                .setFileName(fileName)
                .setHostname(hostname)
                .setFileSize(fileSize)
                .setIp(ip)
                .build();
        NameNodeModel.InformReceivedReplicateResponse response = blockingStub.informReceivedReplicate(request);
    }

    /**
     * 向自己负责通信的那个NameNode进行注册
     */
    public boolean register() throws InterruptedException {
        boolean existsFileInfoOnNameNode = false;
        try {
            // 通过RPC接口发送到NameNode他的注册接口上去
            NameNodeModel.RegisterRequest registerRequest = NameNodeModel.RegisterRequest.newBuilder()
                    .setIp(DataNodeConfig.DATANODE_IP)
                    .setHostname(DataNodeConfig.DATANODE_HOSTNAME)
                    .setPort(DataNodeConfig.DATANODE_PORT)
                    .build();
            // 发送rpc接口调用请求到NameNode去进行注册
            NameNodeModel.RegisterResponse registerResponse = blockingStub.register(registerRequest);
            int status = registerResponse.getStatus();
            if (status == 1) {
                System.out.println(DataNodeConfig.DATANODE_IP + "  " + DataNodeConfig.DATANODE_HOSTNAME + "  " + DataNodeConfig.FILE_DATA + "注册成功，服务端返回状态码为：" + status);
            } else {
                System.out.println(DataNodeConfig.DATANODE_IP + "  " + DataNodeConfig.DATANODE_HOSTNAME + "注册失败，服务端返回状态码为：" + status);
                existsFileInfoOnNameNode = true;
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return existsFileInfoOnNameNode;
    }

    public void heartBeat() {
        HeartbeatThread heartbeatThread = new HeartbeatThread();
        heartbeatThread.start();
    }


    public void reportCompleteFileInfo() {
        JSONArray jsonArray = new JSONArray();
        AtomicLong storeFileSize = new AtomicLong(0);
        // 获取该节点下的所有文件以及文件大小总和
        File dir = new File(DataNodeConfig.FILE_DATA);
        File[] files = dir.listFiles();
        if (files != null || files.length > 0) {
            for (File file : files) {
               scanFiles(file, jsonArray, dir.getAbsolutePath(), storeFileSize);
            }
        }
        System.out.println("文件大小为：" + storeFileSize.longValue());
        NameNodeModel.ReportCompleteFileRequest request = NameNodeModel.ReportCompleteFileRequest.newBuilder()
                .setHostname(DataNodeConfig.DATANODE_HOSTNAME)
                .setIp(DataNodeConfig.DATANODE_IP)
                .setFilenames(jsonArray.toJSONString())
                .setFileStoreSize(storeFileSize.longValue())
                .build();
        NameNodeModel.ReportCompleteFileResponse response = blockingStub.reportCompleteFileInfo(request);
        System.out.println("全量上报后返回结果为：" + response.getStatus());
    }

    private void scanFiles(File dir, JSONArray jsonArray, String parentDirPath, AtomicLong storeFileSize) {
        if (dir.isFile()) {
            String absolutePath = dir.getAbsolutePath();
            String path = absolutePath.replace(parentDirPath, "").replaceAll("\\\\", "/");
            jsonArray.add(path);
            storeFileSize.addAndGet(dir.length());
            System.out.println("文件路径为：" + path + "文件大小为：" + storeFileSize);
        } else {
            File[] files = dir.listFiles();
            if (files != null && files.length > 0) {
                for (File file : files) {
                    scanFiles(file, jsonArray, parentDirPath, storeFileSize);
                }
            }
        }
    }


    /**
     * 负责注册的线程
     *
     * @author zhonghuashishan
     */

    class HeartbeatThread extends Thread {
        @Override
        public void run() {
            try {
                while (true) {
                    // 通过RPC接口发送到NameNode他的注册接口上去
                    NameNodeModel.HeartbeatRequest heartbeatRequest = NameNodeModel.HeartbeatRequest.newBuilder()
                            .setIp(DataNodeConfig.DATANODE_IP)
                            .setHostname(DataNodeConfig.DATANODE_HOSTNAME)
                            .build();

                    // 发送rpc接口调用请求到NameNode去进行注册
                    NameNodeModel.HeartbeatResponse heartbeatResponse = blockingStub.heartbeat(heartbeatRequest);
                    int status = heartbeatResponse.getStatus();
                    if (status == 8) {
                        // 进行任务拷贝
                    }
                    if (status == 1) {
                        System.out.println(DataNodeConfig.DATANODE_IP + "  " + DataNodeConfig.DATANODE_HOSTNAME + "发送心跳成功，服务端返回状态码为：" + status);
                    } else if (status == 4) {
                        // 需要重新注册
                        register();
                        // 重新全量上报文件
                        reportCompleteFileInfo();
                    } else {
                        System.out.println(DataNodeConfig.DATANODE_IP + "  " + DataNodeConfig.DATANODE_HOSTNAME + "发送心跳失败，服务端返回状态码为：" + status);
                    }
                    Thread.sleep(30 * 1000);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


}
